package com.zk.storm;
//每日系统平均点击量
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;

public class DailyPlatformAvgBolt extends BaseRichBolt {
    private OutputCollector collector;
    private Map<String, Long> platformTotalMap;  // 总点击量
    private Map<String, Integer> platformCountMap;  // 记录数

    @Override
    public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        this.platformTotalMap = new HashMap<>();
        this.platformCountMap = new HashMap<>();
    }

    @Override
    public void execute(Tuple input) {
        try {
            String dateKey = input.getStringByField("date_key");
            String platform = input.getStringByField("platform");
            int hits = input.getIntegerByField("hits");

            String key = dateKey + "#" + platform;
            platformTotalMap.put(key, platformTotalMap.getOrDefault(key, 0L) + hits);
            platformCountMap.put(key, platformCountMap.getOrDefault(key, 0) + 1);

            double avgHits = (double) platformTotalMap.get(key) / platformCountMap.get(key);
            collector.emit(new Values(key, avgHits));
            collector.ack(input);
        } catch (Exception e) {
            collector.fail(input);
            throw new RuntimeException("系统平均点击量聚合失败: " + e.getMessage());
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("agg_key", "avg_hits"));  // agg_key=date#platform
    }
}
